package cn.itcast.xc.analysis.visit

import cn.itcast.xc.common.EtlEnvironment
import org.apache.spark.sql.SparkSession

/**
 * <p>
 * 课程访问量统计分析程序
 * </p>
 **/
object CourseVisitProcessByDay {

  // 获取spark session
  val spark: SparkSession = EtlEnvironment.getSparkSession(this.getClass.getSimpleName)

  def main(args: Array[String]): Unit = {

    val date_info =args(0)
//    val date_info = "2019-11-11"
    // 1, 获取访问量事实表数据
    val visitDF = spark.sql(s"select concat_ws('-',years,months,days) as date_info, years, months, days from data_course.course_visit_fact where concat_ws('-',years,months,days)='${date_info}'")
    // 方便测试不过滤日期， 统计所有时间
//    val visitDF = spark.sql(s"select concat_ws('-',years,months,days) as date_info, years, months, days from data_course.course_visit_fact")

    // 2， 求和
    val visitCount = visitDF.groupBy("date_info", "years", "months", "days").count()
    // 2019-11-11 | 2019 | 11 | 11 | 123
    // 求和结果查看
    visitCount.show()

    // 3， 保存到数据仓库（业务表中）
    // 配置允许动态分区
    spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

    visitCount
      // 指定列，防止数据混乱
      .selectExpr("date_info", "count", "years", "months", "days")
      // 写入模式
      .write.mode("overwrite")
      // 指定数据仓库中的表名（库名.表名）
      .insertInto("data_course.course_visit_biz")

    // 数据仓库中的结果查看
    spark.sql("select * from data_course.course_visit_biz").show()

    // 4， 关闭资源
    spark.close()

  }

}
